package peerToPeerFacebook.proj;

import java.io.BufferedReader;
import java.io.BufferedWriter;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.util.HashSet;
import java.util.Set;

/** Class that implements File system APIs. APIs are implemented in transactional manner and each instance of this calls
 * represents separate transaction. This class has no idea about network and deals only with local files. 2PC is
 * implemented in DataManagerCoordinator class. 
 * 
 */
public class DataManager implements IServerFileIO
{
	protected enum LogRecords
	{
		/** Read lock obtained, param is file name */
		ReadLock,
		
		/** Write lock obtained. Dirty file created. Param is original file name */
		WriteLock,
		
		/** Pre-Commit marker. Replay has to make sure not to abort transaction after this point */
		PreCommit,
		
		/** Finalyzing of transaction started. There is not end message - missing log file indicates that. Param is one
		 * of teh states either commit or Abort */
		FlushStart,
		
		/** File has been flushed - param is file name */
		FlushingFile,
		
		/** This is not action name, this is param to FlushStart. Indicates that transaction should be commited */
		Commit,
		
		/** This is not action name, this is param to FlushStart. Indicates that transaction should be aborted */
		Abort
	}
	
	public enum TransactionStatus
	{
		/** Transaction is started and running. */
		Running,
		
		/** Transaction is ion pre-commit state - this means that is ready to be committed and will not abort
		 * automatically in the failure. It will reject any operations except commit or abort.
		 */
		PreCommit,
		
		/** Transaction is finished and committed */
		Comitted,
		
		/** Transaction is finished and aborted */
		Aborted,
		
		/** Commit or abort failed, thus this object is broken and not usable. The only way to continue is to restart
		 * transaction. This can happen only if some undetected file system error occurs or due to bug.
		 */
		Broken
	}
		
	/** Storage class that should be used for file input/output operations */
	protected IStorageProvider storage;
	
	/** ID of currently active transaction that this manager is running */
	protected String transactionID = null;
	
	/** Coordinator used to manage all active transactions */
	protected LockManager locks = null;
	
	/** Transaction log file that should be used for this transaction */
	protected Log log = null;

	/** Set of files for which we have read lock */
	protected Set<String> readFiles = new HashSet<String>();
	
	/** Set of files for which we have write lock */
	protected Set<String> writeFiles = new HashSet<String>();
	
	/** Status of current transaction */
	protected TransactionStatus status = null;

	public DataManager(String transactionID, IStorageProvider storage, LockManager locks) throws IOException
	{
		this.storage = storage;
		this.transactionID = transactionID;
		this.locks = locks;
		this.status = TransactionStatus.Running;
		
		this.log = new Log(storage, getLogName(transactionID));
		
		// Check if we need to restore state. 
		if (storage.fileExists(getLogName(transactionID)))
			restoreTransaction();
	}
	
	/** Returns if log for transaction ID is present. The fact that log is present also menas that the tarsnaction is
	 * active and should be restored.
	 *   
	 * @param storage Storage to use
	 * @param transactionID Transaction ID
	 * @return True if transaction log exists.
	 */
	public static boolean logFileExists(IStorageProvider storage, String transactionID)
	{
	    return storage.fileExists(getLogName(transactionID));
	}
	
	/** Returns transaction ID associated with this data manger.
	 * @return Transaction ID.
	 */
	public String transactionID()
	{
		return transactionID;
	}

	/** Method that creates a new empty file and throws FileAlreadyExistException
	 * if file is already there
	 * 
	 * @param name The name of the file
	 */
	public void createFile(String name) throws IOException
	{
		ensureInTransaction();
		
		// First check if the file exists
		if (storage.fileExists(getFileName(name)))
			throw new FileAlreadyExistException(name);
		
		// Now we need to obtain write lock. Obtaining a write lock does not guarantee that the file exists, so we need
		// to make sure that we create it too.
		ensureFileWrite(name);
		BufferedWriter writer = storage.getWriter(getFileName(name), false);
		writer.close();
	}
	
	/** This method deletes the requested file
	 * 
	 * @param name The file which needs to be deleted
	 * @return true If file gets deleted Or false if doesn't
	 */
	public boolean deleteFile(String name) throws IOException
	{
		ensureInTransaction();
		if (!storage.fileExists(getFileName(name)))
			throw new FileNotFoundException();

		ensureFileWrite(name);

		// If this is the fist time we are getting write lock for this file, then the file be re-created after the lock
		// is obtained. Otherwise the file would be there already as we would have thrown exception. So we always need
		// to delete the file. 
		return storage.delete(getFileName(name));
	}

	/** This method returns the contents of the file specified in a String
	 * 
	 * @param name The file requested
	 * @return Contents of the file Or throws FileNotFoundException otherwise
	 */
	public String getFileContents(String name) throws IOException
	{
		ensureInTransaction();
		if (!storage.fileExists(getFileName(name)))
			throw new FileNotFoundException();

		ensureFileRead(name);
		
		// Read the content
		BufferedReader reader = null;
		StringBuilder result = new StringBuilder();
		try
		{
			reader = storage.getReader(getFileName(name));

			int r = 0;
			char[] buf = new char[1024];
			while ((r = reader.read(buf)) != -1)
			{
				result.append(buf, 0, r);
			}
		} finally
		{
			if (reader != null)
				reader.close();
		}

		return result.toString();
	}

	/** Method that writes the contents to the file mentioned
	 * 
	 * @param name The file specified
	 * @param contents The contents that needs to be written to the file
	 */
	public void putContentsToFile(String name, String contents) throws IOException
	{
		ensureInTransaction();
		writeToFile(name, contents, false);
	}

	/** Appends the contents to the end of the file.
	 * 
	 * @param name The file mentioned
	 * @param contents The contents that needs to be appended
	 */
	public void appendContentsToFile(String name, String contents) throws IOException
	{
		ensureInTransaction();
		writeToFile(name, contents, true);
	}
	
	/** Writes to the file given content.
	 * 
	 * @param name Name of the file
	 * @param contents Content to write
	 * @param append Should content be appended
	 */
	protected void writeToFile(String name, String contents, boolean append) throws IOException
	{
		// If the file does not exist, throw FileNotFoundException
		if (!storage.fileExists(getFileName(name)))
			throw new FileNotFoundException();

		ensureFileWrite(name);
		
		// Flush data
		BufferedWriter file = null; 
		try
		{
			file = storage.getWriter(getFileName(name), append);
			file.write(contents);
		} finally
		{
			if (file != null)
				file.close();
		}
	}
	
	/** Returns status of the transaction.
	 * @return Status of the transaction.
	 */
	public TransactionStatus getSatus()
	{
		return status;
	}
	
	/** Moves transaction to pre-commit state. This means that transaction is ready to be commited and will not
	 * automatically abort if crash happens. The only valid operation after this one is commit and abort.
	 */
	public void preCommit() throws IOException
	{
		ensureInTransaction();
		log.writeMark(LogRecords.PreCommit.toString(), null);
		status = TransactionStatus.PreCommit;
	}

	/** Commits the transaction.
	 * After this is called this object is no longer usable. Failure during commit will tear down this object and
	 * should not be used anymore. 
	 */
	public void commit() throws IOException
	{
		assert status == TransactionStatus.PreCommit: "Transaction needs to be in pre-comit mode before it can be " +
				"commited. This is actually optional so if needed code can be changed to drop this requirement";
		finish(true, true);
	}
	
	/** Aborts the transaction.
	 * After this is called this object is no longer usable. Failure during commit will tear down this object and
	 * should not be used anymore. 
	 */
	public void abort() throws IOException
	{
		assert status == TransactionStatus.PreCommit || status == TransactionStatus.Running: "Transaction should be " +
				"either running or in pre-commit mode to abort it.";
		finish(false, true);
	}
	
	/** Performs either commit or abort. The only difference between those two operations is dirty files are restored
	 * 	to original ones. After commit transaction object is no longer usable
	 * 
	 * @param commit Should transaction be committed
	 * @param writeHeader Should flushStart be written 
	 */
	protected void finish(boolean commit, boolean writeHeader) throws IOException
	{
		// If we fail during commit - this object would not be usable anymore. Mark so
		status = TransactionStatus.Broken;
		
		// As always first log
		// We cheat a bit, technically this is start - end pair. But Log does not support that
		// so we use marks to achieve the same thing
		if (writeHeader)
			log.writeMark(LogRecords.FlushStart.toString(), 
					commit ? LogRecords.Commit.toString() : LogRecords.Abort.toString());
		
		// First deal with files that we have write lock for
		for (String name: writeFiles)
		{
			log.writeStart(LogRecords.FlushingFile.toString(), name);
			// If we are rolling back - there is nothing to do
			if (commit)
				copyFile(getFileName(name), name);
			
			// clean up old dirty files
			if (storage.fileExists(getFileName(name)))
				storage.delete(getFileName(name));
			
			locks.releaseLock(transactionID, name);
			log.writeEnd(LogRecords.FlushingFile.toString());
			
			// This file was in read locks too, let's make sure to drop it
			readFiles.remove(name);
		}
		
		writeFiles = null;
		
		// We are done with writes, now reads. It is interesting to note that for reads we do not need to write any
		// marks. Just release the lock.
		for (String name: readFiles)
			locks.releaseLock(transactionID, name);
		
		readFiles = null;
		
		// Hey we are done - get rod of the log file
		// We do not need mark as if the log is deleted it is already mark. If not, the log will be replayed and then
		// deleted
		storage.delete(getLogName(transactionID));
		log = null;
		
		status = commit ? TransactionStatus.Comitted : TransactionStatus.Aborted;
	}
	
	/** Restores partially finished transaction. This could mean that tarnsaction gets aborted if we can.
	 */
	protected void restoreTransaction() throws IOException
	{
		// This variable is valid iff transaction is in broken state
		boolean commit = false;
		boolean terminate = false;
		Log.LogRecord lastRecord = null;
		
		// read log records and see what thet say		
		for (Log.LogRecord record: log.read())
		{
			assert !terminate: "Log replay should not have finished now!";
			lastRecord = record;
		
			// If action is not completed we need to termiante.
			terminate = !record.completed;
			switch(LogRecords.valueOf(record.action))
			{
				case ReadLock:
					ensureInTransaction();
					if (record.completed)
					{
						boolean got = locks.tryReadLock(transactionID, record.param);
						assert got: "Failed to obtain lock that we should have gotten";
						readFiles.add(record.param);
					}
					break;
				case WriteLock:
					ensureInTransaction();
					if (record.completed)
					{
						boolean got = locks.tryWriteLock(transactionID, record.param);
						assert got: "Failed to obtain lock that we should have gotten";
						readFiles.add(record.param);
						writeFiles.add(record.param);
					} else {
						// the reocrd is not complete - we need to revert any bak files created
						if (storage.fileExists(getFileName(record.param)))
							storage.delete(getFileName(record.param));
					}
					break;
				case PreCommit:
					assert record.completed: "This is martk record, should always be complete";
					assert status == TransactionStatus.Running: "PreCommit is valid only in running status";
					status = TransactionStatus.PreCommit;
					break;
				case FlushStart:
					assert record.completed: "Finish start action is always complete";
					assert LogRecords.valueOf(record.param) == LogRecords.Commit ||
							LogRecords.valueOf(record.param) == LogRecords.Abort:
								"Finish param in the log can only be commit or abort";
					
					status = TransactionStatus.Broken;
					commit = LogRecords.valueOf(record.param) == LogRecords.Commit ? true : false;
					break;
				case FlushingFile:
					assert status == TransactionStatus.Broken: "Flushing files can be done only when transaction " + 
							"enters commit or abort";
					
					// if it is not completed we will need to finish it later. Otherwise we can clean up our state
					if (record.completed)
					{
						readFiles.remove(record.param);
						writeFiles.remove(record.param);
						
						// release the lock
						locks.releaseLock(transactionID, record.param);
					}
					break;
				default:
					assert false: "Unrecognized command: " + record.action;
					break;
			}
		}
		
		// At this point we finished reading the log and can be in several states.
		switch(status)
		{
			case Running:
				// if we are in running mode we can just abort. The only thing to worry is that record may not be
				// completed
				if (!lastRecord.completed)
					log.writeIgnore(lastRecord.action);

				abort();
				break;
			case PreCommit:
				// There is nothing to do in this state. All records are fine and someone needs to tell as what to do
				// next
				break;
			case Broken:
				// in general we can just replay the commit or abot transaction. our in memory context is the same
				// as it was just before the crash.
				// However, if last record was file flush and it is not complete, we need to make sure that we ignore
				// this record.
				if (lastRecord.action.equals(LogRecords.FlushingFile.toString()) && !lastRecord.completed)
					log.writeIgnore(lastRecord.action);
				
				// We already have file information with the transaction commit/abor info so we need to skip it
				finish(commit, false);
				break;
			default:
				assert false: "Unsuported transaction state";
				break;		
		}
	}

	/** Ensures that file is ready for read operation. If this is first read then read lock is obtained. If such 
	 * cannot be obtained, the IOEXception is thrown. Finally if the lock is obtained mark is written in the log
	 * indicating that such thing happened.
	 * 
	 * @param file Name of the file that is about to be read
	 */
	protected void ensureFileRead(String file) throws IOException
	{
		// If we already have the file opened for read there is nothing to do
		if (readFiles.contains(file))
			return;
		
		// Otherwise we need to do some housekeeping
		if (!locks.tryReadLock(transactionID, file))
			throw new IOException("Failed to obtain read lock.");			
		
		// TODO: It seems that we can write that we got the lock AFTER it was got. But this needs to be revisited.
		log.writeMark(LogRecords.ReadLock.toString(), file);
		readFiles.add(file);
	}
	
	/** Ensures that the file is ready to be written to. If this is the first write then write lock is obtained, copy
	 * of the file is made and log is updated.
	 * 
	 * @param file Name of the file
	 */
	protected void ensureFileWrite(String file) throws IOException
	{
		// If we already have the file opened for read there is nothing to do
		if (writeFiles.contains(file))
			return;
		
		// Otherwise we need to do some housekeeping. We get both locks as at this time we do both operations
		if (!locks.tryWriteLock(transactionID, file) || !locks.tryReadLock(transactionID, file))
			throw new IOException("Failed to obtain write lock.");			
		
		// TODO: It seems that we can write that we got the lock AFTER it was got. But this needs to be revisited.
		// As this is first write, we need to make a copy of it
		log.writeStart(LogRecords.WriteLock.toString(), file);
		copyFile(file, dirtyFileName(file));
		log.writeEnd(LogRecords.WriteLock.toString());
		
		// At this point we can write to file too, so we updated both lists
		readFiles.add(file);
		writeFiles.add(file);
	}
	
	/** Asserts if transaction has been finished */
	protected void ensureInTransaction()
	{
		assert status == TransactionStatus.Running: "Log is null meaning that transaction either ended or never started";
	}

	/** Returns file name that should be used for now. Basically returns file name if the file is not opened for
	 * writing and dirty name of the file if the file is opened for writing.
	 * 
	 * @param name Original name of the file
	 * @return Filename that should be used for operation.
	 */
	protected String getFileName(String name)
	{
		if (writeFiles.contains(name))
			return dirtyFileName(name);
		else
			return name;
	}
	
	/** Makes copy if the file. If the source file is missing then destination is deleted. Otherwise destination file
	 * is over written.
	 * 
	 * @param from Filename to copy
	 * @param to Filename to copy to
	 */
	protected void copyFile(String from, String to) throws IOException
	{
		// First deal with the case where the file needs to be simply be deleted
		if (!storage.fileExists(from))
		{
			// Destination is there so we need to clean up to make sure that they are the same
			if (storage.fileExists(to))
				storage.delete(to);
			
			// If the file is not there - we are done.
			return;
		}
		
		// Now we know that we really need to move some context around
		BufferedReader source = storage.getReader(from);
		BufferedWriter destination = storage.getWriter(to, false);
		try
		{
			String line = null;
			while ((line = source.readLine()) != null)
			{
				destination.write(line);
				destination.newLine();
			}
		} finally
		{
			if (source != null)
				source.close();

			if (destination != null)
				destination.close();
		}
	}
	
	/** Returns dirty file name
	 * @param originalName Original file name
	 * @return Dirty file name
	 */
	protected String dirtyFileName(String originalName)
	{
		return originalName + ".dirty";
	}

	/** Returns transaction log name
	 * @param transactionID Transaction ID
	 * @return Transaction log name
	 */
	protected static String getLogName(String transactionID)
	{
		return "dm-log-" + transactionID;
	}
}
